package cn._51doit.live.jobs;

import cn._51doit.live.pojo.DataBean;
import cn._51doit.live.udfs.IsNewUserFunction;
import cn._51doit.live.udfs.JsonToBeanFunction;
import cn._51doit.live.utils.EventType;
import cn._51doit.live.utils.FlinkUtils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

/**
 * 实时统计统计新老用户（从今天凌晨开始，到当前）
 * 数据中没有isNew这个字段了，需要根据设备ID进行计算
 */
public class NewUserCountV3 {

    public static void main(String[] args) throws Exception {

        //从指定的文件中读取参数
        ParameterTool parameterTool = ParameterTool.fromPropertiesFile(args[0]);

        DataStream<String> lines = FlinkUtils.createKafkaStream(parameterTool, SimpleStringSchema.class);

        FlinkUtils.env.setStateBackend(new EmbeddedRocksDBStateBackend(true)); //true代表开启增量checkpoint
        //指定checkpopint数据在HDFS中存储的路径
        FlinkUtils.env.getCheckpointConfig().setCheckpointStorage(parameterTool.getRequired("checkpoint.path"));


        //将数据进行转换（字符串转成Bean）
        SingleOutputStreamOperator<DataBean> beanStream = lines.process(new JsonToBeanFunction());

        //过滤数据类型
        SingleOutputStreamOperator<DataBean> filtered = beanStream.filter(bean -> EventType.APP_LAUNCH.equals(bean.getEventId()));

        //按照设备ID进行keyBy，这样可以保证设备ID相同的数据一定进入到同一个分区
        //存在的问题：按照设备ID进行keyBy，会导致存储的key非常多，存储的状态会非常多
        //解决方案（使用RocksDB StateBacked、或HashMapStateBackend）
        //咱们使用RocksDB StateBacked，因为可以进行增量checkpoint，提高checkpoint的速度
        KeyedStream<DataBean, String> keyedStream = filtered.keyBy(DataBean::getDeviceId);

        SingleOutputStreamOperator<DataBean> res = keyedStream.process(new IsNewUserFunction());

        res.print();

        FlinkUtils.env.execute();

    }
}
